
                /*********************************************
                **                                          **
                **                                          **
                **                g.d.nb.cmr1.C             **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/



/*
** Direct Access Global and Non-Blocking Global Read/Write Functions 
** Inter Group communication
** even number of processes
*/

#include "arch.h"
#include "A.utilities"

//user program

Global_id gid1, gid2;
Coll_id cid1, cid2, cid3;
Coll_id cid4, cid5, cid6;

//on most homogenous systems, explicit setting 
//won't be necessary (try commenting the
//setting lines, you should get the same values
//for the cids).
void init(int, char **){
   cid1.set();
   cid2.set();
   cid3.set();
   cid4.set();
   cid5.set();
   cid6.set();
   gid1.set();
   gid2.set();
   fprintf(Sdf::trace, "node%d: cid1=%d, cid2=%d, cid3=%d, cid4=%d, cid5=%d, cid6=%d, gid1=%d, gid2=%d\n",
           Sdf::PROC_ID, 
           cid1.get(), cid2.get(), cid3.get(),
           cid4.get(), cid5.get(), cid6.get(),
           gid1.get(), gid2.get());
   fflush(Sdf::trace);
}

int count;

void wcs_handler(HRW_rcd *ptr){
   count--;
   Sdf::Access_type access_type;
   int tag_dv = ptr->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d: global-writing's over on call side\n", Sdf::PROC_ID);
   fprintf(Sdf::trace, "node%d: wcs_handler, rem_proc=%d, d_ptr=%d, g-sz=%d, th_ptr=%d, hndlr_ptr=%d, hndlr_d_ptr=%d, tag_dv=%d, atype=%d, tag_av=%d, tag_ev=%d, comm=%d, offset=%d, Tsize=%d\n",
           Sdf::PROC_ID, 
           ptr->get_remote_proc(), 
           ptr->get_data_ptr(), 
           ptr->get_given_size(), 
           ptr->get_thread_ptr(),
           ptr->get_hndlr_ptr(), 
           ptr->get_hndlr_data_ptr(),
           tag_dv,
           access_type,
           ptr->get_tag_av(),
           ptr->get_tag_ev(),
           ptr->get_comm(),
           ptr->get_offset(),
           ptr->get_Tsize());
   fflush(Sdf::trace);

}

void rcs_handler(HRW_rcd *ptr){
   count--;
   Sdf::Access_type access_type;
   int tag_dv = ptr->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d: global-reading's over on call side\n", Sdf::PROC_ID);
   fprintf(Sdf::trace, "node%d: rcs_handler, rem_proc=%d, d_ptr=%d, g-sz=%d, r_sz=%d, th_ptr=%d, hndlr_ptr=%d, hndlr_d_ptr=%d, tag_dv=%d, atype=%d, tag_av=%d, tag_ev=%d, comm=%d, offset=%d, Tsize=%d\n",
           Sdf::PROC_ID,
           ptr->get_remote_proc(), 
           ptr->get_data_ptr(),
           ptr->get_given_size(), 
           ptr->get_rcvd_size(),
           ptr->get_thread_ptr(), 
           ptr->get_hndlr_ptr(),
           ptr->get_hndlr_data_ptr(),
           tag_dv,
           access_type,
           ptr->get_tag_av(),
           ptr->get_tag_ev(),
           ptr->get_comm(),
           ptr->get_offset(),
           ptr->get_Tsize());
   fflush(Sdf::trace);
}


class Root_proc: public Thread{

   public:

        Root_proc()
                : Thread((Tmf0_ptr)&Root_proc::body){}

        void body(){

                fprintf(Sdf::trace, "node %d, in Root_proc::start\n", Sdf::PROC_ID);
                fflush(Sdf::trace); 

                const int N = Sdf::PROC_NB/2;
                if(Sdf::PROC_NB%2){
                   printf("node %d, !!!!!execute with even process number: terminated\n", Sdf::PROC_ID);
                   fflush(stdout);
                   return;
                }

                Group *group1;
                int *rk1 = new int[N];
                int i;
                for(i = 0; i < N; i++)
                   rk1[i] = i;
                group1 = Sdf::incl(Sdf::COMML_WORLD, N, rk1);
                CommL *comm1 = group1->commit(cid1, Sdf::COMML_WORLD);

                Group *group2;
                int *rk2 = new int[N];
                for(i = N; i < Sdf::PROC_NB; i++)
                   rk2[i-N] = i;
                group2 = Sdf::incl(Sdf::COMML_WORLD, Sdf::PROC_NB - N, rk2);
                CommL *comm2 = group2->commit(cid2, Sdf::COMML_WORLD);

                Sdf::barrier(cid4);

                CommR *commr;
                commr = Sdf::newCommR(cid3, comm1, comm2, Sdf::COMML_WORLD);

                int from = Sdf::PROC_ID;
                int to1 = 777;
                int to2 = 888;

                Global<int> global1(gid1, &to1, commr);
                Global<int> global2(gid2, &from, commr);
                Sdf::barrier(cid5, commr);                
                  //synchronizes global construction over commr

                {Sdf::Access_type access_type;
                 int tag_dv = global1.get_tag_dv(&access_type);
                 fprintf(Sdf::trace, "node%d, test global1 accessors, tag_ev=%d, tag_dv=%d, access_type=%d, tag_av=%d, direct_handle=%d, comm=%d\n",
                         Sdf::PROC_ID, 
                         global1.get_tag_ev(),
                         tag_dv,
                         access_type,
                         global1.get_tag_av(),
                         global1.get_direct_handle(),
                         global1.get_comm());
                 fflush(Sdf::trace);
                }
                {Sdf::Access_type access_type;
                 int tag_dv = global2.get_tag_dv(&access_type);
                 fprintf(Sdf::trace, "node%d, test global2 accessors, tag_ev=%d, tag_dv=%d, access_type=%d, tag_av=%d, direct_handle=%d, comm=%d\n",
                         Sdf::PROC_ID, 
                         global2.get_tag_ev(),
                         tag_dv,
                         access_type,
                         global2.get_tag_av(),
                         global2.get_direct_handle(),
                         global2.get_comm());
                 fflush(Sdf::trace);
                }

                  //each process in local group write into same 
                  //rank process in remote group.
                int rank = commr->get_local_CommL()->rank();
                count = 0;
                global1.write(rank, 0, 
                              from, 1, wcs_handler, 0, 0);
                count++;
                global2.read(rank, 0, 
                             to2, 1, rcs_handler, 0, 0);
                count++;
                while(count)
                   Sdf::poll();
                       //achieves local completion

                Sdf::barrier(cid6);
                       //achieves global completion

                fprintf(Sdf::trace, "node %d, from=%d, to1=%d, to2=%d\n", 
                        Sdf::PROC_ID, from, to1, to2);
                fflush(Sdf::trace);

                delete []rk1;
                delete []rk2;

                delete comm1;
                delete comm2;
                delete commr;

                fprintf(Sdf::trace, "node %d, the end\n", Sdf::PROC_ID);
                fflush(Sdf::trace);
        }
};
//end of user program

MAIN(Root_proc, init, Sdf::empty, Sdf::empty)


